/*
 * Copyright (C) 2010 Fores Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package jp.fores.midori.client.handler;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.concurrent.CountDownLatch;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import jp.fores.midori.client.RPCClient;
import jp.fores.midori.core.dto.RPCInputDto;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpException;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.HttpVersion;
import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.impl.nio.DefaultClientIOEventDispatch;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.NHttpConnection;
import org.apache.http.nio.entity.NByteArrayEntity;
import org.apache.http.nio.protocol.BufferingHttpClientHandler;
import org.apache.http.nio.protocol.EventListener;
import org.apache.http.nio.protocol.HttpRequestExecutionHandler;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.SessionRequest;
import org.apache.http.nio.reactor.SessionRequestCallback;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.CoreConnectionPNames;
import org.apache.http.params.HttpParams;
import org.apache.http.params.HttpProtocolParams;
import org.apache.http.protocol.BasicHttpProcessor;
import org.apache.http.protocol.ExecutionContext;
import org.apache.http.protocol.HttpContext;
import org.apache.http.protocol.HttpRequestExecutor;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestTargetHost;

/**
 * リモートプロシージャーコール用を行うクライアントの通信処理を行うハンドラのHttpCore-NIOを使った実装クラス。<br>
 * <a href="http://hc.apache.org/index.html">Apache HttpComponents</a>の <code>{@link ConnectingIOReactor}</code> クラスを使って処理を行います。<br>
 * (このクラスの全てのメソッドはスレッドセーフです。)<br>
 */
public class HttpCoreNioRPCHandler implements RPCHandler {
	//==========================================================
	//定数

	/**
	 * ログ出力用
	 */
	private static final Log log = LogFactory.getLog(HttpCoreNioRPCHandler.class);

	/**
	 * バッファサイズ
	 */
	private static final int BUFFER_SIZE = 8192;


	//==========================================================
	//フィールド

	//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
	//クラスフィールド

	/**
	 * HttpProcessor
	 */
	//(スレッドセーフなクラスなので、クラスフィールドとして定義して共有する)
	private static final BasicHttpProcessor httpProcessor = new BasicHttpProcessor();

	/**
	 * HttpRequestExecutor
	 */
	//(スレッドセーフなクラスなので、クラスフィールドとして定義して共有する)
	private static final HttpRequestExecutor httpRequestExecutor = new HttpRequestExecutor();


	//==========================================================
	//メソッド

	//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
	//staticイニシャライザ
	static {
		//HttpProcessorに各種Interceptorを設定
		httpProcessor.addInterceptor(new RequestContent());
		httpProcessor.addInterceptor(new RequestTargetHost());
		httpProcessor.addInterceptor(new RequestConnControl());

		//「User-Agent」ヘッダは自分で設定するのでコメントアウト
		//httpProcessor.addInterceptor(new RequestUserAgent());

		//「Expect: 100-Continue handshake」機能は使わないのでコメントアウト
		//httpProcessor.addInterceptor(new RequestExpectContinue());
	}


	//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
	//抽象メソッドの実装

	/**
	 * {@inheritDoc}
	 */
	@Override
	public Object execute(RPCClient rpcClient, URL url, RPCInputDto inputDto)
			throws Exception {
		//==========================================================
		//リクエストの設定

		//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
		//URLをHttpHostに変換

		//URLのポートを取得
		int port = url.getPort();

		//ポートを取得できなかった場合
		if (port == -1) {
			//代わりに「80」を使う
			port = 80;
		}

		//URLのホストとポートを元にして、HttpHostのインスタンスを生成
		HttpHost host = new HttpHost(url.getHost(), port);


		//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
		//パラメーターの設定
		//(TODO:設定方法が本当にあっているかもう少し確認する)

		//パラメータークラスのインスタンスを生成
		HttpParams params = new BasicHttpParams();

		params.setIntParameter(CoreConnectionPNames.SOCKET_BUFFER_SIZE, 8 * 1024);
		params.setBooleanParameter(CoreConnectionPNames.STALE_CONNECTION_CHECK, false);
		params.setBooleanParameter(CoreConnectionPNames.TCP_NODELAY, true);

		//HTTPのバージョンに「1.1」を設定
		HttpProtocolParams.setVersion(params, HttpVersion.HTTP_1_1);

		//文字コードに「UTF-8」を設定
		HttpProtocolParams.setContentCharset(params, "UTF-8");

		//「Expect: 100-Continue handshake」機能は使わないので無効にする
		HttpProtocolParams.setUseExpectContinue(params, false);

		//接続タイムアウト時間を設定
		params.setIntParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, rpcClient.connectTimeout);

		//読み取りタイムアウト時間を設定
		params.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, rpcClient.readTimeout);


		//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
		final ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(2, params);

		CountDownLatch requestCount = new CountDownLatch(1);

		try {

			MyHttpRequestExecutionHandler executionHandler = new MyHttpRequestExecutionHandler(requestCount, rpcClient, inputDto, url);

			BufferingHttpClientHandler handler = new BufferingHttpClientHandler(httpProcessor, executionHandler, new DefaultConnectionReuseStrategy(), params);

			handler.setEventListener(new EventLogger());

			final IOEventDispatch ioEventDispatch = new DefaultClientIOEventDispatch(handler, params);

			Thread t = new Thread(new Runnable() {

				public void run() {
					try {
						ioReactor.execute(ioEventDispatch);
					}
					catch (InterruptedIOException ex) {
						//デバッグログが有効な場合
						if (log.isDebugEnabled()) {
							//デバッグログ出力
							log.debug("Interrupted");
						}
					}
					catch (IOException e) {
						//デバッグログが有効な場合
						if (log.isDebugEnabled()) {
							//デバッグログ出力
							log.debug("I/O error", e);
						}
					}

					//デバッグログが有効な場合
					if (log.isDebugEnabled()) {
						//デバッグログ出力
						log.debug("スレッド終了");
					}
				}

			});
			t.start();

			SessionRequest sessionRequest = ioReactor.connect(new InetSocketAddress(host.getHostName(), host.getPort()), null, host, new MySessionRequestCallback(requestCount));

			// Block until all connections signal
			// completion of the request execution
			requestCount.await();

			Exception exception = executionHandler.getException();

			if (exception != null) {
				throw exception;
			}

			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("正常終了");
			}

			return executionHandler.getResult();
		}
		finally {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Shutting down I/O reactor");
			}

			ioReactor.shutdown(1);

			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Shut down success");
			}
		}

	}

	private static class MyHttpRequestExecutionHandler implements
			HttpRequestExecutionHandler {

		private final static String REQUEST_SENT = "request-sent";

		private final static String RESPONSE_RECEIVED = "response-received";


		//==========================================================
		//フィールド

		private final CountDownLatch requestCount;

		/**
		 * リモートプロシージャーコール用を行うクライアント
		 */
		private final RPCClient rpcClient;

		/**
		 * 入力データ用のDTO
		 */
		private final RPCInputDto inputDto;

		/**
		 * URL
		 */
		private final URL url;

		/**
		 * 戻り値
		 */
		private Object result = null;

		/**
		 * 例外
		 */
		private Exception exception = null;


		//==========================================================
		//メソッド

		public MyHttpRequestExecutionHandler(final CountDownLatch requestCount,
				RPCClient rpcClient, RPCInputDto inputDto, URL url) {
			super();
			this.requestCount = requestCount;

			//リモートプロシージャーコール用を行うクライアント
			this.rpcClient = rpcClient;

			//入力データ用のDTO
			this.inputDto = inputDto;

			this.url = url;
		}

		public void initalizeContext(final HttpContext context,
				final Object attachment) {
			HttpHost targetHost = (HttpHost) attachment;
			context.setAttribute(ExecutionContext.HTTP_TARGET_HOST, targetHost);
		}

		public void finalizeContext(final HttpContext context) {
			Object flag = context.getAttribute(RESPONSE_RECEIVED);
			if (flag == null) {
				// Signal completion of the request execution
				this.requestCount.countDown();
			}
		}

		public HttpRequest submitRequest(final HttpContext context) {
			HttpHost targetHost = (HttpHost) context.getAttribute(ExecutionContext.HTTP_TARGET_HOST);
			Object flag = context.getAttribute(REQUEST_SENT);
			if (flag == null) {
				context.setAttribute(REQUEST_SENT, Boolean.TRUE);

				try {

					//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
					//リクエストメソッドに応じたクラスを使用する

					//リクエストメソッドを大文字にして取得
					String requestMethod = this.rpcClient.requestMethod.toUpperCase();

					//リクエスト用のクラス
					//(setEntity()ができないとサーバー側にデータを送れないので、型はHttpEntityEnclosingRequestBaseに限定する)
					final HttpEntityEnclosingRequestBase request;

					//リクエストメソッドが「POST」の場合
					if (HttpPost.METHOD_NAME.equals(requestMethod)) {
						//POST用のリクエストクラスのインスタンスを使用する
						request = new HttpPost(this.url.toURI());
					}
					//リクエストメソッドが「PUT」の場合
					else if (HttpPut.METHOD_NAME.equals(requestMethod)) {
						//PUT用のリクエストクラスのインスタンスを使用する
						request = new HttpPut(this.url.toURI());
					}
					//それ以外の場合
					else {
						//どうしようもないので、不正な引数の場合の例外を投げる
						//(メッセージには大文字に変換する前のリクエストメソッドを含める)
						throw new IllegalArgumentException("リクエストメソッドが不正です。:"
								+ this.rpcClient.requestMethod);
					}

					//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
					//リクエストヘッダの設定

					//「User-Agent」ヘッダに自分の単純なクラス名を設定
					//(this.getClass()を使うと内部クラスの名前になってしまうので、クラスを直接指定する)
					request.addHeader("User-Agent", HttpCoreNioRPCHandler.class.getSimpleName());

					//「Content-Type」ヘッダを設定
					//(rpcClient本体ではなく、rpcClientの中のserializerから取得する)
					request.addHeader("Content-Type", this.rpcClient.serializer.getContentType());

					//リクエスト送信時にgzip圧縮を行う場合
					if (this.rpcClient.requestGZip) {
						//「Content-Encoding」用のヘッダを設定
						request.addHeader(this.rpcClient.getContentEncodingHeaderName(), "gzip");
					}

					//レスポンスでgzip圧縮を受け入れる場合
					if (this.rpcClient.responseGZip) {
						//「Accept-Encoding」用のヘッダを設定
						request.addHeader(this.rpcClient.getAcceptEncodingHeaderName(), "gzip,deflate");
					}


					//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
					//サーバーに入力データを渡す
					ByteArrayOutputStream bos = new ByteArrayOutputStream();

					OutputStream outputStream = bos;

					try {
						//リクエスト送信時にgzip圧縮を行う場合
						if (this.rpcClient.requestGZip) {
							//デバッグログが有効な場合
							if (log.isDebugEnabled()) {
								//デバッグログ出力
								log.debug("リクエスト送信時にgzip圧縮を行います。");
							}

							//出力ストリームをGZIPOutputStreamでラッピングする
							outputStream = new GZIPOutputStream(outputStream);
						}
						//それ以外の場合
						else {
							//出力ストリームをBufferedOutputStreamでラッピングする
							outputStream = new BufferedOutputStream(outputStream, BUFFER_SIZE);
						}

						this.rpcClient.serializer.inputDtoToOutputStream(this.inputDto, outputStream);
					}
					//終了処理
					finally {
						//出力ストリームを確実に閉じる
						//(ここで出力ストリームを閉じなくても、入力ストリームを開こうとすると自動的に閉じられるようだが、念のために閉じておく)
						if (outputStream != null) {
							try {
								outputStream.close();
							}
							//例外処理
							catch (Exception e) {
								//どうしようもないので、例外はもみ消す
							}
						}
					}

					//入力データ用のDTOの内容を出力ストリーム経由でEntityに出力するためのクラスを使う
					//(標準で用意されているEntityでは直接出力ストリームには出力できないため)
					//request.setEntity(new EntityTemplate(new InternalContentProducer(this.rpcClient, this.inputDto)));
					//request.setEntity(new BufferingNHttpEntity(new EntityTemplate(new InternalContentProducer(this.rpcClient, this.inputDto)), new HeapByteBufferAllocator()));
					request.setEntity(new NByteArrayEntity(bos.toByteArray()));


					//デバッグログが有効な場合
					if (log.isDebugEnabled()) {
						//デバッグログ出力
						log.debug("リクエスト作成完了");
					}
					//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
					return request;
				}
				catch (Exception e) {
					//デバッグログが有効な場合
					if (log.isDebugEnabled()) {
						//デバッグログ出力
						log.debug("リクエスト作成失敗", e);
					}

					return null;
				}
			}
			else {
				// No new request to submit
				return null;
			}
		}

		public void handleResponse(final HttpResponse response,
				final HttpContext context) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("handleResponse");
			}

			context.setAttribute(RESPONSE_RECEIVED, Boolean.TRUE);

			HttpHost targetHost = (HttpHost) context.getAttribute(ExecutionContext.HTTP_TARGET_HOST);
			try {
				//レスポンスのEntityを取得
				HttpEntity responseEntity = response.getEntity();

				//デバッグログが有効な場合
				if (log.isDebugEnabled()) {
					//デバッグログ出力
					log.debug("responseEntity:" + responseEntity);
				}


				//レスポンスのEntityの入力ストリームを取得
				InputStream inputStream = responseEntity.getContent();

				try {
					//レスポンスの「Content-Encoding」用のヘッダの内容を取得
					Header contentEncoding = response.getFirstHeader(this.rpcClient.getContentEncodingHeaderName());

					//レスポンスの「Content-Encoding」用のヘッダに「gzip」という文字列が含まれている場合
					if ((contentEncoding != null)
							&& (contentEncoding.getValue().indexOf("gzip") != -1)) {
						//デバッグログが有効な場合
						if (log.isDebugEnabled()) {
							//デバッグログ出力
							log.debug("レスポンスがgzip圧縮されているので、展開しつつ読み込みます。");
						}

						//入力ストリームをGZIPInputStreamでラッピングする
						inputStream = new GZIPInputStream(inputStream);
					}
					//それ以外の場合
					else {
						//入力ストリームをBufferedInputStreamでラッピングする
						inputStream = new BufferedInputStream(inputStream, BUFFER_SIZE);
					}

					//レスポンスの入力ストリームを処理して、リモートプロシージャコールの戻り値を取得
					//(rpcClientの中のserializerに処理を委譲する)
					Object result = this.rpcClient.serializer.handleRPCResponseInputStream(inputStream);

					//デバッグログが有効な場合
					if (log.isDebugEnabled()) {
						//デバッグログ出力
						log.debug("*** 「" + targetHost
								+ "」に対するリモートプロシージャコールが正常終了しました ***");
					}

					//リモートプロシージャコールの戻り値を返す
					this.result = result;
				}
				//終了処理
				finally {
					//入力ストリームを確実に閉じる
					if (inputStream != null) {
						try {
							inputStream.close();
						}
						//例外処理
						catch (Exception e) {
							//どうしようもないので、例外はもみ消す
						}
					}
				}
			}
			catch (Exception e) {
				this.exception = e;
			}

			// Signal completion of the request execution
			this.requestCount.countDown();
		}

		public Object getResult() {
			return this.result;
		}

		public Exception getException() {
			return this.exception;
		}
	}

	private static class MySessionRequestCallback implements
			SessionRequestCallback {

		private final CountDownLatch requestCount;

		public MySessionRequestCallback(final CountDownLatch requestCount) {
			super();
			this.requestCount = requestCount;
		}

		public void cancelled(final SessionRequest request) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Connect request cancelled: "
						+ request.getRemoteAddress());
			}

			this.requestCount.countDown();
		}

		public void completed(final SessionRequest request) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("completed");
			}
		}

		public void failed(final SessionRequest request) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Connect request failed: "
						+ request.getRemoteAddress());
			}

			this.requestCount.countDown();
		}

		public void timeout(final SessionRequest request) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Connect request timed out: "
						+ request.getRemoteAddress());
			}
			this.requestCount.countDown();
		}

	}

	private static class EventLogger implements EventListener {

		public void connectionOpen(final NHttpConnection conn) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Connection open: " + conn);
			}
		}

		public void connectionTimeout(final NHttpConnection conn) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Connection timed out: " + conn);
			}
		}

		public void connectionClosed(final NHttpConnection conn) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("Connection closed: " + conn);
			}
		}

		public void fatalIOException(final IOException ex,
				final NHttpConnection conn) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("I/O error: " + ex.getMessage());
			}
		}

		public void fatalProtocolException(final HttpException ex,
				final NHttpConnection conn) {
			//デバッグログが有効な場合
			if (log.isDebugEnabled()) {
				//デバッグログ出力
				log.debug("HTTP error: " + ex.getMessage());
			}
		}

	}
}
